package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.arithmetic.common.Cn6BasicDataInfo;
import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.offline.service.LoadCn6BasicDataInfoService;
import com.navinfo.tripanalysis.offline.service.TripEngineConvertService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

@RunningType(value = RunningTypeEnum.HS, interfaceClass = LoadCn6BasicDataInfoService.class )
public class LoadCn6BasicDataInfoServiceImpl implements LoadCn6BasicDataInfoService {
    private String dbName;

    private String tbName;

    private String selectHql;

    public String getDbName() {
        return dbName;
    }

    public void setDbName(String dbName) {
        this.dbName = dbName;
    }

    public String getTbName() {
        return tbName;
    }

    public void setTbName(String tbName) {
        this.tbName = tbName;
    }

    public String getSelectHql() {
        return selectHql;
    }

    public void setSelectHql(String selectHql) {
        this.selectHql = selectHql;
    }
    @Override
    public Broadcast<HashMap<Long, Cn6BasicDataInfo>> getCn6BasicDataInfoRdd(SparkSession spark, JavaSparkContext jsc) {

        String infoHql = String.format(this.selectHql, this.dbName, this.tbName);

        Dataset<Row> dataSet = spark.sql(infoHql);

        JavaPairRDD<Long, Cn6BasicDataInfo> cn6BasicDataInfoRDD = dataSet.toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, Cn6BasicDataInfo>) (Iterator<Row> rowIterator) -> {
            List<Tuple2<Long, Cn6BasicDataInfo>> cn6BasicDataInfoList = new ArrayList<>();
            while (rowIterator.hasNext()) {
                Row row = rowIterator.next();
                Cn6BasicDataInfo cn6BasicDataInfo = new Cn6BasicDataInfo();
                cn6BasicDataInfo.setVin(row.getAs("vin"));
                cn6BasicDataInfo.setNoticeModel(row.getAs("noticeModel"));
                cn6BasicDataInfo.setEmissionValue(row.getAs("emissionValue"));
                cn6BasicDataInfo.setTerminalNo(Long.valueOf(row.getAs("terminalNo")));
                cn6BasicDataInfo.setEngineModelName(row.getAs("engineModelName"));
                cn6BasicDataInfo.setEngineModelCode(row.getAs("engineModelCode"));
//                cn6BasicDataInfo.setTorqueMax(row.getAs("torqueMax"));
//                cn6BasicDataInfo.setCyclePower(row.getAs("cyclePower"));
//                cn6BasicDataInfo.setCityType(row.getAs("cityType"));
                cn6BasicDataInfoList.add(new Tuple2(cn6BasicDataInfo.getTerminalNo(), cn6BasicDataInfo));
            }
            return cn6BasicDataInfoList.iterator();
        });
        return jsc.broadcast(new HashMap<>(cn6BasicDataInfoRDD.collectAsMap()));

    }

}
